package com.tech.common

import com.tech.config.ApplicationConfig
import kafka.producer.{KeyedMessage, Producer, ProducerConfig}

import java.util.Properties

object KafkaUtil {

  val props = new Properties()
  props.put("metadata.broker.list", ApplicationConfig.KAFKA_BROKER_LIST) // broker 如果有多个,中间使用逗号分隔
  props.put("serializer.class", "kafka.serializer.StringEncoder")
  props.put("request.required.acks", "-1")
  val config = new ProducerConfig(props)
  @transient lazy val producer = new Producer[String, String](config)

  def write2Kafka(topic: String, msg: String): Unit = {
    val producerRecord = new KeyedMessage[String, String](topic, msg)
    producer.send(producerRecord)
  }

  def write2KafkaWithKey(topic: String, key: String, msg: String): Unit = {
    val producerRecord = new KeyedMessage[String, String](topic, key, msg)
    producer.send(producerRecord)
  }
}
